package org.apache.kafka.niezhipeng.customSerializerDemo.derializer;

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.niezhipeng.customSerializerDemo.pojo.Company;
import org.apache.kafka.niezhipeng.tools.ByteReader;

import java.nio.charset.StandardCharsets;

/**
 * @author Nie ZhiPeng
 * @date 2025/7/25 0:14
 */
public class CompanyDerializer implements Deserializer<Company> {
    @Override
    public Company deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        Company company = new Company();
        if (data.length > 0) {
            ByteReader byteReader = new ByteReader(data);
            int nameLength = byteReader.readInt();
            byte[] nameBytes = byteReader.read(nameLength);
            String name = new String(nameBytes, StandardCharsets.UTF_8);

            int addressLength = byteReader.readInt();
            byte[] addressBytes = byteReader.read(addressLength);
            String address = new String(addressBytes, StandardCharsets.UTF_8);

            company.setName(name);
            company.setAddress(address);
        }
        return company;
    }
}
